package com.venus.kafka.producer.service;

import java.util.Map;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import com.google.common.collect.Maps;
import com.google.gson.Gson;
import com.venus.kafka.bean.MessageInfo;

import lombok.extern.slf4j.Slf4j;

/**
 * @author lwl
 * @date 2020/9/1 17:58
 */
@Service
@Slf4j
public class ProducerService {

	private static final String TOPIC = "venus";

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	public void sendMessage() {
		MessageInfo messageInfo = new MessageInfo();

		messageInfo.setId("11111");
		messageInfo.setMessage("just try it");
		Map<String, String> attrs = Maps.newHashMap();
		attrs.put("11", "11");
		attrs.put("22", "22");
		attrs.put("33", "33");
		Gson gson = new Gson();
		log.info("#########start send message : " + attrs.toString());
		// 发送消息，订单类的 json 作为消息体
		ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(TOPIC, gson.toJson(messageInfo));

		// 监听回调
		future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

			@Override
			public void onFailure(Throwable throwable) {
				log.info("## Send message fail ...");
			}

			@Override
			public void onSuccess(SendResult<String, String> result) {
				log.info("## Send message success ...");
			}
		});
	}

}
